package com.bds.btcdc.adapter.okex.market;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bds.btcdc.api.model.Result;
import com.bds.btcdc.model.*;
import com.bds.btcdc.util.Const;
import com.bds.btcdc.util.Utils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompSession;

import java.net.URI;
import java.nio.ByteBuffer;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * Okex的WebSocket客戶端，接受okex市场深度更新推送
 */
public class OkexDepthReceiver extends WebSocketClient {
	private Logger logger = LoggerFactory.getLogger(getClass());
	private static final DecimalFormat df = new DecimalFormat("#0.00");
	private static final DecimalFormat df2 = new DecimalFormat("#0.0000");

	StompSession session;
	List<CoinPairExchanges> okex_pairs;

	public OkexDepthReceiver(StompSession session, URI serverURI, List<CoinPairExchanges> okex_pairs) {
		super(serverURI);
		this.session = session;
		this.okex_pairs = okex_pairs;
	}

	public OkexDepthReceiver(URI serverUri, Draft draft) {
		super(serverUri, draft);
	}

	public OkexDepthReceiver(URI serverURI) {
		super(serverURI);
	}

	@Override
	public void onOpen(ServerHandshake handshakedata) {
		String dataMsg = "{'event':'addChannel','channel':'ok_sub_spot_btc_usdt_depth_5'}";
	/*	for (int i = 0; i < okex_pairs.size(); i++) {
			String pair = okex_pairs.get(i).getPair_code();
			dataMsg += "{'event':'addChannel','channel':'ok_sub_spot_" + pair + "_depth_5'},";
		}
		dataMsg.substring(0, dataMsg.length() - 1);
		dataMsg = "[" + dataMsg + "]";
		// System.out.println(dataMsg);*/
		send(dataMsg);
		logger.info("######### Okex 连接成功 #############");
	}


	@Override
	public void onClose(int code, String reason, boolean remote) {
		logger.info("######### Okex连接断开，重连中 ######## 断开原因：" + reason);
		new Thread() {
			@Override
			public void run() {
				try {
					Thread.sleep(Const.RE_CONN_TIME);
					new OkexDepthReceiver(session, new URI(Const.OKEX.WS_URL), okex_pairs).connect();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}.start();
	}

	@Override
	public synchronized void onMessage(String message) {
//		logger.info("okex深度更新>>>\n" + message);
		Result rst = new Result();
		String json = Utils.regex("\\[(.*)\\]", message);
		JSONObject json_object = JSONObject.parseObject(json);
		String channel = (String) json_object.get("channel");
		if("addChannel".equals(channel))
			return;
		// 取channel参数中的币种名称
		String pair = Utils.regex("spot_(.+)_depth", channel);
		JSONObject data=json_object.getJSONObject("data");
		JSONArray asks=data.getJSONArray("asks");
		JSONArray bids=data.getJSONArray("bids");

		MarketDepth depth=new MarketDepth();
		depth.setExchange(Const.OKEX.EXCHANGE);
		depth.setPair(pair);

		List<Order> bid_list=new ArrayList<>();
		for (int i = 0; i <bids.size(); i++) {
			JSONArray array=bids.getJSONArray(i);
			Order order=new Order();
			order.setPrice(df.format(array.getBigDecimal(0)));
			order.setAmount(df2.format(array.getBigDecimal(1)));
			bid_list.add(order);
		}

		List<Order> ask_list=new ArrayList<>();
		for (int i = 0; i <asks.size(); i++) {
			JSONArray array=asks.getJSONArray(i);

			Order order=new Order();
			order.setPrice(df.format(array.getBigDecimal(0)));
			order.setAmount(df2.format(array.getBigDecimal(1)));
			ask_list.add(order);
		}
		depth.setBids(bid_list);
		depth.setAsks(ask_list);

		rst.setCode(Const.SUC);
		rst.setData(depth);

		if (session != null) {
			// 推到btc-web主题
			try {
				synchronized (session) {
					session.send(Const.OKEX.MARKET_TOPIC, JSONObject.toJSONString(rst).getBytes());
				}
			} catch (Exception e) {
				logger.error("", e);
			}
		}
	}

	@Override
	public void onMessage(ByteBuffer message) {
		logger.info("received ByteBuffer");
	}

	@Override
	public void onError(Exception ex) {
		logger.info("an error occurred:" + ex);
	}

	public static void main(String[] args) throws Exception {
		new OkexDepthReceiver(new URI("wss://real.okex.com:10440/websocket/okexapi")).connect();
	}
}